package com.mars.kafkademo.controller;

import com.mars.kafkademo.interceptor.CountInterceptor;
import com.mars.kafkademo.interceptor.TimeInterceptor;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * author by: mars
 * Date: 2020/4/12 17:39
 * Description: 生产者
 */
public class CustomProducer {

    public static void main(String[] args) {
        Properties props = new Properties();
        // kafka服务端的ip:port 可替换成 主机名:port
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.5.115:9092");
        // 等待所有副本节点的应答
        props.put(ProducerConfig.ACKS_CONFIG,"all");
        // 消息发送最大尝试次数
        props.put(ProducerConfig.RETRIES_CONFIG,0);
        // 一批消息处理大小
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
        // 请求延时
        props.put(ProducerConfig.LINGER_MS_CONFIG,1);
        // 发送缓存区内存大小
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
        // key序列化,注意是kafkad的序列化包 org.apache.kafka.common.serialization.StringSerializer
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // value序列化
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        // 注册生产者的拦截器
        List<String> interceptorList = new ArrayList<>();
        // 顺序链拦截
        interceptorList.add(TimeInterceptor.class.getName());
        interceptorList.add(CountInterceptor.class.getName());
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,interceptorList);

        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(props);
        for (int i=0;i<50;i++){
            // 1.send()的返回对象是 ：Future<RecordMetadata>
            producer.send(new ProducerRecord<String,String>("mars11",Integer.toString(i),"hello-world-"+i));
            // 2.send()带回调函数
            producer.send(new ProducerRecord<String, String>("mars11", Integer.toString(i), "hello-world-" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // 回调 recordMetadata 与 e 是互斥的，当发送成功，e=null，发送失败，recordMetadata=null
                    if (recordMetadata!=null) {
                        System.out.println("发送成功");
                    }
                }
            });
            // 3.send() 指定partition后，key hash不生效
            producer.send(new ProducerRecord<>("mars11",0,"aaa",""));
            // 4.send() 不指定partition，key按照hash，根据创建的分区进行hash，发送消息
            producer.send(new ProducerRecord<>("mars11","aaa",""));
        }
        producer.close();
    }

}
